package cn.gwm.flink.streaming.task;


import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.factory.DataHubSourceFactory;
import cn.gwm.flink.streaming.function.process.OriginalDataProcessFunction;
import cn.gwm.flink.streaming.ods.ModelHandle;
import cn.gwm.flink.streaming.ods.model.ModelCanEnum;
import cn.gwm.flink.streaming.ods.model.StandardFieldConf;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.utils.ConfigLoader;
import cn.gwm.utils.DateTimeUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.FixedLinkedHashMap;
import cn.hutool.db.Db;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.shaded.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class OdsEs11Task {

    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        // 3.1）设置每隔30s一个周期开启checkpoint（checkpoint开启周期，不能太长也不能太短：1-5分钟）
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        setEnv(env);
//        Configuration configuration = new Configuration();
//        configuration.setInteger(RestOptions.PORT, 8081);
//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
//        env.setParallelism(1);

        Map<String, OutputTag<String>> canTagMap = new ConcurrentHashMap<>(16);
        for (ModelCanEnum canEnum : ModelCanEnum.values()) {
            canTagMap.put(canEnum.getModelName(), new OutputTag<String>(canEnum.getValue()) {});
        }

        handle(env, canTagMap);


        env.execute("OdsEs11Task");
    }

    private static void handle(StreamExecutionEnvironment env, Map<String, OutputTag<String>> canTagMap) {
        SingleOutputStreamOperator<String> process =  env.addSource(DataHubSourceFactory.getDatahubSourceFunction("otsp_pdcu_fault_detection_data_dev",
                        "1685432499178CNBQ6", DateTimeUtil.defaultFormatPare(ConfigLoader.get("all.receive.endTime"))))
                .process(new OriginalDataProcessFunction(canTagMap)).returns(Types.STRING);
        CleanData cleanData = new CleanData();
        SingleOutputStreamOperator<JSONObject> operator = process.getSideOutput(canTagMap.get(ModelCanEnum.ES11.getModelName()))
                .flatMap(cleanData).returns(JSONObject.class).name("flatMapEs11");
        SingleOutputStreamOperator<JSONObject> streamOperator = operator.keyBy(item -> item.getStr(BaseFields.vin)).process(new ProcessFunction<JSONObject, JSONObject>() {
            private ValueState<Long> preTimeState;

            @Override
            public void open(Configuration parameters) throws Exception {
                preTimeState = getRuntimeContext().getState(new ValueStateDescriptor<>("preTimeState", Long.class));
            }

            @Override
            public void processElement(JSONObject entries, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                Long preTime = preTimeState.value();
                if (preTime == null) {
                    preTimeState.update(entries.getLong(BaseFields.tid));
                    out.collect(entries);
                } else {
                    if (preTime < entries.getLong(BaseFields.tid)) {
                        preTimeState.update(entries.getLong(BaseFields.tid));
                        out.collect(entries);
                    }
                }
            }
        });
        /*SingleOutputStreamOperator<JSONObject> apply = operator.keyBy(item -> item.getStr(BaseFields.vin))
//                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
//                        .withTimestampAssigner((element, recordTimestamp) -> element.getLong(BaseFields.tid)))
//                .keyBy(item -> item.getStr(BaseFields.vin))
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.minutes(15)))
                .apply(new WindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<JSONObject> input, Collector<JSONObject> out) throws Exception {
                        Map<String,JSONObject> map = new TreeMap<>(new Comparator<String>() {
                            @Override
                            public int compare(String o1, String o2) {
                                return o1.compareTo(o2);
                            }
                        });
                        Iterator<JSONObject> iterator = input.iterator();
                        while (iterator.hasNext()) {
                            JSONObject next = iterator.next();
                            map.put(next.getStr(BaseFields.tid)+next.getStr(BaseFields.command),next);
                        }
                        for(Map.Entry<String,JSONObject> entry: map.entrySet()){
                            out.collect(entry.getValue());
                        }

                    }
                });*/
        OutputTag<JSONObject> gpsTag = new OutputTag<JSONObject>("gpsTag"){};
        OutputTag<JSONObject> gpsHbaseTag = new OutputTag<JSONObject>("gpsHbaseTag"){};
        OutputTag<JSONObject> oneTag = new OutputTag<JSONObject>("oneTag"){};
        OutputTag<JSONObject> twoTag = new OutputTag<JSONObject>("twoTag"){};
        OutputTag<JSONObject> threeTag = new OutputTag<JSONObject>("threeTag"){};
//        SingleOutputStreamOperator<JSONObject> stream = operator.keyBy(item -> item.getStr(BaseFields.vin))
//                .process(new BloomFilterProcessFunction(gpsTag,gpsHbaseTag))
//                .name("process_es11_dedup").uid("process_es11_dedup");
        SingleOutputStreamOperator<JSONObject> stream = streamOperator.keyBy(item -> item.getStr(BaseFields.vin))
                .process(new SideOutDataProcessFunction(gpsTag,gpsHbaseTag,oneTag,twoTag,threeTag))
                .name("process_es11_dedup").uid("process_es11_dedup");
        /*stream.flatMap(new FlatMapFunction<JSONObject, Object>() {
            @Override
            public void flatMap(JSONObject object, Collector<Object> collector) throws Exception {
                print(object);
            }
        });
        DataStream<JSONObject> gpsStream = stream.getSideOutput(gpsTag);
        gpsStream.flatMap(new FlatMapFunction<JSONObject, Object>() {
            @Override
            public void flatMap(JSONObject object, Collector<Object> collector) throws Exception {
                print(object);
            }
        });*/
       stream.addSink(FlinkKafkaUtil.toKafka("ods_otsp_es11_60")).name("toKafkaDefault_es11")
                .uid("ods_toKafkaDefault_id_es11");
//        operator.print();
        stream.addSink(BaseTask.getOrcTxtSink("ods_can_es11","ods/otsp/can/60", cleanData.getList()))
                .name("toCanHive_es11").uid("ods_toCanHive_id_es11");
        DataStream<JSONObject> gpsStream = stream.getSideOutput(gpsTag);
        gpsStream.addSink(BaseTask.getOrcTxtSink("ods_gps_es11","ods/otsp/gps/60",
                        BeanSource.SourceEnum.gpsIntranet)).name("toGPSHive_es11")
                .uid("ods_toGpsHive_id_es11");
        DataStream<JSONObject> hbaseStream = stream.getSideOutput(gpsHbaseTag);
        hbaseStream.addSink(BaseTask.getOrcTxtSink("ods_gps_es11_b","ods/otsp/gps/60",
                        BeanSource.SourceEnum.gpsIntranet)).name("toGPSHive_es11_hbase")
                .uid("ods_toGpsHive_id_es11_hbase");
        stream.getSideOutput(threeTag).addSink(FlinkKafkaUtil.toKafka("ods_otsp_es11_45")).name("toKafkaDefault_es11_45")
                .uid("ods_toKafkaDefault_id_es11_45");
        stream.getSideOutput(threeTag).addSink(BaseTask.getOrcTxtSink("ods_can_es11","ods/otsp/can/45", cleanData.getList()))
                .name("toCanHive_es11_45").uid("ods_toCanHive_id_es11_45");
        stream.getSideOutput(twoTag).addSink(FlinkKafkaUtil.toKafka("ods_otsp_es11_30")).name("toKafkaDefault_es11_30")
                .uid("ods_toKafkaDefault_id_es11_30");
        stream.getSideOutput(twoTag).addSink(BaseTask.getOrcTxtSink("ods_can_es11","ods/otsp/can/30", cleanData.getList()))
                .name("toCanHive_es11_30").uid("ods_toCanHive_id_es11_30");
        stream.getSideOutput(oneTag).addSink(FlinkKafkaUtil.toKafka("ods_otsp_es11_15")).name("toKafkaDefault_es11_15")
                .uid("ods_toKafkaDefault_id_es11_15");
        stream.getSideOutput(oneTag).addSink(BaseTask.getOrcTxtSink("ods_can_es11","ods/otsp/can/15", cleanData.getList()))
                .name("toCanHive_es11_15").uid("ods_toCanHive_id_es11_15");



//        operator.addSink(BaseTask.getOrcTxtSink("ods_can_es11","ods/can/", BeanSource.SourceEnum.evOdsTh)).name("toHiveEs11");
//        operator.addSink(FlinkKafkaUtil.toKafka("ods_can_status_es11_temp")).name("toKafkaDefaultEs11");
    }

    private static void print(JSONObject object) {
//        if("LGWEETA50MK075883".equals(object.getStr(BaseFields.vin))) {
            System.out.println(object.getStr(BaseFields.vin) + " : " + object.getLong(BaseFields.tid)
                    + " : " + DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
                    DatePattern.NORM_DATETIME_PATTERN) + " : " + object.getStr(BaseFields.command));
//        }
    }

    private static void print12(JSONObject object) {
        if("LGWEETA50MK075883".equals(object.getStr(BaseFields.vin))){
            System.out.println(object.getStr(BaseFields.vin)+" : "+object.getLong(BaseFields.tid)
                    +" : "+DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
                    DatePattern.NORM_DATETIME_PATTERN)+" : "+object.getStr(BaseFields.command) +" 重复===");
        }
    }

    private static void setEnv(StreamExecutionEnvironment env) {
        env.enableCheckpointing(300 * 1000L);
        // 启动类型
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        // 3.2）设置检查点的mode，exactly-once，保证数据的一次性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 3.3）设置两次checkpoint之间的时间间隔，避免两次间隔时间太近导致频繁的checkpoint
        env.getCheckpointConfig().setCheckpointInterval(150*1000L);
        // 3.4）设置checkpoint的超时时间（一般设置一个checkpoint周期的二分之一）
        env.getCheckpointConfig().setCheckpointTimeout(300 * 1000L);
        // 3.5）设置checkpoint的最大尝试次数，同一个时间内有几个checkpoint在运行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        // 3.6）设置job作业取消的时候，是否保留checkpoint的计算结果
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 3.7）设置job作业运行过程中，如果checkpoint失败，是否停止作业
        env.getCheckpointConfig().setFailOnCheckpointingErrors(true);
        // 3.8）设置checkpoint的存储后端，使用rocksdb作为状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        // 设置可容忍的检查点的失败数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        //启用不对齐检查点保存方式
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        env.getCheckpointConfig().setCheckpointStorage(ConfigLoader.get("hdfsUri") + "/flink/checkpoints/" + "OdsEs11Task");
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.minutes(15),Time.seconds(10)));
    }
    static class CleanData implements FlatMapFunction<String, JSONObject> {
        List<String> list;
        Map<String,String> codeMap;
        public CleanData(){
            initList();
        }
        private void initList() {
            this.list = new ArrayList<>();
            this.codeMap = new ConcurrentHashMap<>(16);
            try {
                Db.use(DSFactory.get(ConfigLoader.get("mysql.group.db.config")))
                        .find(StandardFieldConf.getEntityWhere(), StandardFieldConf.class).stream()
                        .filter(data -> ModelHandle.BIZ_TYPE_ODS.equals(data.getBizType())
                                &&Integer.parseInt(data.getState())>0
                                &&(("1".equals(data.getVersion())
                                &&"ES11".equals(data.getVehicletype()))
                                ||"BASE".equals(data.getVehicletype())))
                        .sorted(new Comparator<StandardFieldConf>() {
                            @Override
                            public int compare(StandardFieldConf o1, StandardFieldConf o2) {
                                return Integer.parseInt(o1.getId())-Integer.parseInt(o2.getId());
                            }
                        })
                        .forEach(data -> {
                            list.add(data.getSemaphore());
                            codeMap.put(data.getSubCode(),data.getSemaphore());
                        });
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        }
        public  List<String>  getList(){
            return this.list;
        }
        @Override
        public void flatMap(String str, Collector<JSONObject> collector) throws Exception {
            JSONObject object = null;
            try {
                object = JSONUtil.parseObj(str);
            } catch (Exception e) {
                e.printStackTrace();
                throw e;
            }
            //周期can信号
            if(BaseFields.COMMAND_HW_HEARBEAT_DECODED.equals(object.getStr(BaseFields.command))
                    ||BaseFields.COMMAND_HW_LOCATION.equals(object.getStr(BaseFields.command))){
                JSONObject resultOjb = JSONUtil.createObj();
                for (Map.Entry<String,String> entry : codeMap.entrySet()) {
                    if(object.containsKey(entry.getKey())){
                        resultOjb.set(entry.getValue(), object.getStr(entry.getKey()));
                    }
                }
                if(BaseFields.COMMAND_HW_LOCATION.equals(object.getStr(BaseFields.command))){
                    try {
                        if (resultOjb.getDouble(BaseFields.LONGITUDE) > 72
                                && resultOjb.getDouble(BaseFields.LONGITUDE) < 136 && resultOjb.getDouble(BaseFields.LATITUDE) > 3
                                && resultOjb.getDouble(BaseFields.LATITUDE) < 54) {
                            collectData(collector, resultOjb);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        throw e;
                    }
                }else{
                    collectData(collector, resultOjb);
                }
            }
        }
        private void collectData(Collector<JSONObject> collector, JSONObject object) {
            try {
                object.set(BaseFields.STORE_TIME, System.currentTimeMillis())
                        .set(VehicleType.VEHICLE_TYPE, "ES11")
                        .set(BaseFields.PARTITION_KEY, DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
                                DatePattern.NORM_DATE_PATTERN));
            } catch (Exception e) {
                e.printStackTrace();
                throw e;
            }
            object.set(BaseFields.tid,object.getLong(BaseFields.tid)/1000*1000);
            if(object.getLong(BaseFields.tid)>System.currentTimeMillis()-3600000&&object.getLong(BaseFields.tid)<System.currentTimeMillis()+30000){
                collector.collect(object);
            }
        }
    }

    static class BloomFilterProcessFunction extends  ProcessFunction<JSONObject, JSONObject>{

        private static final Logger LOGGER = LoggerFactory.getLogger(BloomFilterProcessFunction.class);
        private OutputTag<JSONObject> gpsTag;
        private OutputTag<JSONObject> gpsHbaseTag;

        private volatile BloomFilter<String> canFilter;
        private volatile BloomFilter<String> gpsFilter;

        private static final int BF_CARDINAL_THRESHOLD =  1000;
        private static final double BF_FALSE_POSITIVE_RATE = 0.01;

        public BloomFilterProcessFunction(OutputTag<JSONObject> gpsTag,OutputTag<JSONObject> gpsHbaseTag){
            this.gpsTag = gpsTag;
            this.gpsHbaseTag = gpsHbaseTag;
        }
        @Override
        public void open(Configuration parameters) throws Exception {
            long s = System.currentTimeMillis();
            canFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), BF_CARDINAL_THRESHOLD, BF_FALSE_POSITIVE_RATE);
            gpsFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), BF_CARDINAL_THRESHOLD, BF_FALSE_POSITIVE_RATE);
            long e = System.currentTimeMillis();
            LOGGER.info("Created Guava BloomFilter, time cost: " + (e - s));
        }
        @Override
        public void processElement(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
            if(BaseFields.COMMAND_HW_HEARBEAT_DECODED.equals(object.getStr(BaseFields.command))) {
                if (!canFilter.mightContain(object.getStr(BaseFields.tid))) {
                    canFilter.put(object.getStr(BaseFields.tid));
                    out.collect(object);
                }
            }else if(BaseFields.COMMAND_HW_LOCATION.equals(object.getStr(BaseFields.command))){
                if (!gpsFilter.mightContain(object.getStr(BaseFields.tid))) {
                    gpsFilter.put(object.getStr(BaseFields.tid));
                   ctx.output(gpsTag,object);
                }
            }
        }
    }

    static class SimpleOutDataProcessFunction extends  ProcessFunction<JSONObject, JSONObject>{
        private OutputTag<JSONObject> gpsTag;
        private OutputTag<JSONObject> canTag;

        public SimpleOutDataProcessFunction(OutputTag<JSONObject> gpsTag,OutputTag<JSONObject> canTag){
            this.gpsTag = gpsTag;
            this.canTag = canTag;
        }

        @Override
        public void processElement(JSONObject value, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
            out.collect(value);
            if(BaseFields.COMMAND_HW_LOCATION.equals(value.getStr(BaseFields.command))){
                ctx.output(gpsTag,value);
            }
            if(BaseFields.COMMAND_HW_HEARBEAT_DECODED.equals(value.getStr(BaseFields.command))){
                ctx.output(canTag,value);
            }
        }
    }

    static class SideOutDataProcessFunction extends ProcessFunction<JSONObject, JSONObject>{
        private static final Logger LOGGER = LoggerFactory.getLogger(SideOutDataProcessFunction.class);
        private OutputTag<JSONObject> gpsTag;
        private OutputTag<JSONObject> gpsHbaseTag;
        private OutputTag<JSONObject> oneTag;
        private OutputTag<JSONObject> twoTag;
        private OutputTag<JSONObject> threeTag;

        private transient MapState<String, JSONObject> mapState;
        //缓存最近32条记录
        private transient ValueState<JSONObject> canObj;
        private transient ValueState<Long> onTimerTs;
        private transient ValueState<FixedLinkedHashMap> fixedMapState;
        private transient ValueState<FixedLinkedHashMap> fixedMapGpsState;

        public SideOutDataProcessFunction(OutputTag<JSONObject> gpsTag,OutputTag<JSONObject> gpsHbaseTag,
                                          OutputTag<JSONObject> oneTag,OutputTag<JSONObject> twoTag,OutputTag<JSONObject> threeTag){
            this.gpsTag = gpsTag;
            this.gpsHbaseTag = gpsHbaseTag;
            this.oneTag = oneTag;
            this.twoTag = twoTag;
            this.threeTag = threeTag;
        }
        @Override
        public void onTimer(long timestamp, ProcessFunction<JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            clearObj(out,ctx);
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            StateTtlConfig ttlConfig = StateTtlConfig
                    .newBuilder(Time.seconds(300))
                    .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired).build();
            MapStateDescriptor<String, JSONObject> mapState = new MapStateDescriptor<>("mapState", String.class, JSONObject.class);
            mapState.enableTimeToLive(ttlConfig);
            this.mapState = getRuntimeContext().getMapState(mapState);
            this.canObj = getRuntimeContext().getState(new ValueStateDescriptor<>("canObj",JSONObject.class));
            this.onTimerTs = getRuntimeContext().getState(new ValueStateDescriptor<>("onTimerTs", Long.class));
            this.fixedMapState = getRuntimeContext().getState(new ValueStateDescriptor<>("fixedMapState", FixedLinkedHashMap.class));
            this.fixedMapGpsState = getRuntimeContext().getState(new ValueStateDescriptor<>("fixedMapGpsState", FixedLinkedHashMap.class));
        }

        @Override
        public void processElement(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
            if(BaseFields.COMMAND_HW_HEARBEAT_DECODED.equals(object.getStr(BaseFields.command))){
//                if(map.contains(object.getStr(BaseFields.tid))){
//                    return;
//                }else{
//                    map.put(object.getStr(BaseFields.tid),object.getStr(BaseFields.vin));
//                    Iterator<Map.Entry<String, String>> iterator = map.iterator();
//                    Map<String,String> sortMap = new TreeMap<>(new Comparator<String>() {
//                        @Override
//                        public int compare(String o1, String o2) {
//                            return o1.compareTo(o2);
//                        }
//                    });
//                    while (iterator.hasNext()){
//                        Map.Entry<String, String> next = iterator.next();
//                        sortMap.put(next.getKey(),next.getValue());
//                    }
//                    if(sortMap.size()>32){
//                        FixedLinkedHashMap linkedHashMap = new FixedLinkedHashMap(32);
//                        linkedHashMap.putAll(sortMap);
//                        map.clear();
//                        map.putAll(linkedHashMap);
//                    }
//                }
//                if (canFilter.mightContain(object.getStr(BaseFields.tid))) {
//                    System.out.println(object.getStr(BaseFields.vin)+" : "+object.getLong(BaseFields.tid)
//                            +" : "+DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
//                            DatePattern.NORM_DATETIME_PATTERN)+" : "+object.getStr(BaseFields.command) +" 重复===");
//                    return;
//                }else{
//                    canFilter.put(object.getStr(BaseFields.tid));
//                }
//                FixedLinkedHashMap fixedMap = fixedMapState.value();
//                if(fixedMap==null){
//                    fixedMap = new FixedLinkedHashMap(128);
//                }
//                if(fixedMap.containsKey(object.getStr(BaseFields.tid))){
//                    print12(object);
//                    return;
//                }else{
//                    fixedMap.put(object.getStr(BaseFields.tid),"1");
//                    fixedMapState.update(fixedMap);
//                }
                if(canObj.value()!=null&&onTimerTs.value()!=null){
                    context.timerService().deleteProcessingTimeTimer(onTimerTs.value());
                    clearObj(out,context);
                }
                if(!mapState.isEmpty()){
                    if(mapState.contains(object.getStr(BaseFields.tid))){
                        JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                        if(obj!=null){
                            extracted(object, obj, 0, out);
                        }else{
                            cacheCanObj(object, context);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                        if(obj!=null){
                            extracted(object, obj, 1, out);
                        }else{
                            cacheCanObj(object, context);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                        if(obj!=null){
                            extracted(object, obj, -1, out);
                        }else{
                            cacheCanObj(object, context);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                        if(obj!=null){
                            extracted(object, obj, 2, out);
                        }else{
                            cacheCanObj(object, context);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                        if(obj!=null){
                            extracted(object, obj, -2, out);
                        }else{
                            cacheCanObj(object, context);
                        }
                    }else{
                        cacheCanObj(object, context);
                    }
                    Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                    while (iterator.hasNext()){
                        Map.Entry<String, JSONObject> next = iterator.next();
                        if( next!=null && next.getKey().compareTo(object.getStr(BaseFields.tid))<=0){
                            iterator.remove();
                        }
                    }
                }else{
                    cacheCanObj(object, context);
                }
            }else if(BaseFields.COMMAND_HW_LOCATION.equals(object.getStr(BaseFields.command))){
//                if(gpsMap.contains(object.getStr(BaseFields.tid))){
//                    return;
//                }else{
//                    gpsMap.put(object.getStr(BaseFields.tid),object.getStr(BaseFields.vin));
//                    Iterator<Map.Entry<String, String>> iterator = gpsMap.iterator();
//                    Map<String,String> sortMap = new TreeMap<>(new Comparator<String>() {
//                        @Override
//                        public int compare(String o1, String o2) {
//                            return o1.compareTo(o2);
//                        }
//                    });
//                    while (iterator.hasNext()){
//                        Map.Entry<String, String> next = iterator.next();
//                        sortMap.put(next.getKey(),next.getValue());
//                    }
//                    if(sortMap.size()>32){
//                        FixedLinkedHashMap linkedHashMap = new FixedLinkedHashMap(32);
//                        linkedHashMap.putAll(sortMap);
//                        map.clear();
//                        map.putAll(linkedHashMap);
//                    }
//                }
//                if (gpsFilter.mightContain(object.getStr(BaseFields.tid))) {
//                    System.out.println(object.getStr(BaseFields.vin)+" : "+object.getLong(BaseFields.tid)
//                            +" : "+DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
//                            DatePattern.NORM_DATETIME_PATTERN)+" : "+object.getStr(BaseFields.command)+" 重复");
//                    return;
//                }else{
//                    gpsFilter.put(object.getStr(BaseFields.tid));
//                }
//                FixedLinkedHashMap fixedMap = fixedMapGpsState.value();
//                if(fixedMap==null){
//                    fixedMap = new FixedLinkedHashMap(128);
//                }
//                if(fixedMap.containsKey(object.getStr(BaseFields.tid))){
//                    print12(object);
//                    return;
//                }else{
//                    fixedMap.put(object.getStr(BaseFields.tid),"1");
//                    fixedMapGpsState.update(fixedMap);
//                }
                //gps信号直接输出到侧输出流
                gpsOutPut(object, context);
                if(canObj!=null&&canObj.value()!=null){
                    JSONObject canObject = canObj.value();
                    if(canObject!=null){
                        if(object.getStr(BaseFields.tid).equals(canObject.getStr(BaseFields.tid))){
                            canObject.set(BaseFields.LONGITUDE, object.getStr(BaseFields.LONGITUDE));
                            canObject.set(BaseFields.LATITUDE, object.getStr(BaseFields.LATITUDE));
                            canObject.set(BaseFields.IS_GPS_FLAG, object.getStr(BaseFields.IS_GPS_FLAG));
                            canObject.set(BaseFields.MATCH_ERROR, 0);
                            out.collect(canObject);
                            canObj.clear();
                            onTimerTs.clear();
                        }else {
                            putGpsObj(object);
                        }
                    }else{
                        putGpsObj(object);
                    }
                }else{
                    putGpsObj(object);
                }
            }
        }

        private void gpsOutPut(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context) {
            context.output(gpsTag, object);
        }

        /**
         * 缓存can周期信号，等待定时器触发
         * @param object
         * @param context
         * @throws IOException
         */
        private void cacheCanObj(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context) throws IOException {
            canObj.update(object);
            long ts = context.timerService().currentProcessingTime() + 8000L;
            onTimerTs.update(ts);
            context.timerService().registerProcessingTimeTimer(ts);
        }

        /**
         *  缓存gps数据
         * @param obj
         * @throws Exception
         */
        private void putGpsObj(JSONObject obj) throws Exception {
            JSONObject object = new JSONObject();
            object.set(BaseFields.LONGITUDE, obj.getStr(BaseFields.LONGITUDE));
            object.set(BaseFields.LATITUDE, obj.getStr(BaseFields.LATITUDE));
            object.set(BaseFields.IS_GPS_FLAG, obj.getStr(BaseFields.IS_GPS_FLAG));
            object.set(BaseFields.tid, obj.getStr(BaseFields.tid));
            object.set(BaseFields.vin, obj.getStr(BaseFields.vin));
            mapState.put(object.getStr(BaseFields.tid), object);
        }

        private void clearObj(Collector<JSONObject> out,Context context) throws Exception {
            if(canObj!=null&&canObj.value()!=null){
                JSONObject object = canObj.value();
                if(object!=null){
                    if(!mapState.isEmpty()){
                        boolean bool = false;
                        if(mapState.contains(object.getStr(BaseFields.tid))){
                            JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                            if(obj!=null) {
                                extracted(object,obj,0,null);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                            if(obj!=null) {
                                extracted(object,obj,1,null);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                            if(obj!=null) {
                                extracted(object,obj,-1,null);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                            if(obj!=null) {
                                extracted(object,obj,2,null);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                            if(obj!=null) {
                                extracted(object,obj,-2,null);
                            }
                        }else{
                            bool = true;
                        }
                        Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                        while (iterator.hasNext()){
                            Map.Entry<String, JSONObject> next = iterator.next();
                            if(next!=null&&bool){
                                bool = false;
                                JSONObject value = next.getValue();
                                context.output(gpsHbaseTag,value);
                            }
                            if( next!=null && (next.getKey().compareTo(object.getStr(BaseFields.tid))<=0)){
                                iterator.remove();
                            }
                        }
                    }
                    out.collect(object);
                    if(object.getLong(BaseFields.tid)>System.currentTimeMillis()-2700000){
                        context.output(threeTag,object);
                    }
                    if(object.getLong(BaseFields.tid)>System.currentTimeMillis()-1800000){
                        context.output(twoTag,object);
                    }
                    if(object.getLong(BaseFields.tid)>System.currentTimeMillis()-900000){
                        context.output(oneTag,object);
                    }
                    canObj.clear();
                    onTimerTs.clear();
                }
            }
        }

        /**
         *  读取gps信号放入can信号中
         * @param object
         * @param obj
         * @param value
         * @param out
         */
        private void extracted(JSONObject object, JSONObject obj, int value, Collector<JSONObject> out) {
            object.set(BaseFields.LONGITUDE, obj.getStr(BaseFields.LONGITUDE));
            object.set(BaseFields.LATITUDE, obj.getStr(BaseFields.LATITUDE));
            object.set(BaseFields.IS_GPS_FLAG, obj.getStr(BaseFields.IS_GPS_FLAG));
            object.set(BaseFields.MATCH_ERROR, value);
            if(null!=out){
                out.collect(object);
            }
        }
    }
}
